package com.jackstraw.jack_base.mq.rocketmq.bean;

import com.jackstraw.jack_base.util.CollectionUtil;
import com.jackstraw.jack_base.util.SpringUtil;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.core.RocketMQTemplate;


import com.jackstraw.jack_base.mq.common.entity.MQEntity;
import com.jackstraw.jack_base.util.ObjectUtil;
import com.jackstraw.jack_base.util.StringUtil;
import com.jackstraw.jack_base.util.IPUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.apache.rocketmq.spring.support.RocketMQUtil;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;

@Data
@Slf4j
public class RocketMqEntity extends MQEntity<RocketMqProducer, RocketMqConsumer, RocketMQTemplateS, RocketMqEntity> {

    public RocketMqEntity() {
        this.setConsumer(new RocketMqConsumer());
    }

    /**
     * accessChannel的Enum类型，取值为LOCAL、CLOUD
     * <p>
     * Enum type for accessChannel, values: LOCAL, CLOUD
     * <p>
     * AccessChannel下枚举值
     */
    private AccessChannel accessChannel;

    @Override
    public boolean checkMQProducer() {
        if (!super.checkMQProducer()) {
            return Boolean.FALSE;
        }
        String group = this.getProducer().getGroup();
        if (StringUtil.StringUtilsBylang3.isEmpty(group)) {
            log.error("ServerName为：{}的mqEntity中[producer.group]为空，请查看配置文件", this.getServerName());
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    @Override
    public boolean checkMQConsumer(RocketMqConsumer mqConsumer) {
        if (!super.checkMQConsumer(mqConsumer)) {
            return Boolean.FALSE;
        }
        String groupName = mqConsumer.getGroup();
        if (StringUtil.StringUtilsBylang3.isEmpty(groupName)) {
            log.warn("ServerName为：{}的mqEntity中[consumer.groupName]为空，请查看配置文件", this.getServerName());
            return Boolean.FALSE;
        }
        String topicName = mqConsumer.getTopic();
        if (StringUtil.StringUtilsBylang3.isEmpty(topicName)) {
            log.warn("ServerName为：{}的mqEntity中[consumer.topicName]为空，请查看配置文件", this.getServerName());
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }

    @Override
    public RocketMQTemplateS mqTemplateS() {
        this.checkMqEntity();
        RocketMQTemplateS rocketMQTemplateS = new RocketMQTemplateS();
        rocketMQTemplateS.setMqEntity(this);
        rocketMQTemplateS.setDefaultMQProducer(this.defaultMQProducer());
        if (this.checkMQConsumer(this.getConsumer())) {
            try {
                rocketMQTemplateS.setDefaultLitePullConsumer(this.defaultConsumer(this.getConsumer()));
            } catch (MQClientException e) {
                log.warn("ServerName为：{}的mqEntity中[consumer]创建，请查看配置文件,异常原因为：{}", this.getServerName(), e);
            }
        }

        RocketMQTemplate rocketMQTemplate = this.rocketMQTemplate(rocketMQTemplateS.getDefaultMQProducer(), rocketMQTemplateS.getDefaultLitePullConsumer());
        rocketMQTemplateS.setRocketMQTemplate(rocketMQTemplate);

        if (CollectionUtil.isNotEmpty(this.getOtherConsumerS())) {
            rocketMQTemplateS.getRocketMQTemplateMap().clear();
            rocketMQTemplateS.getDefaultLitePullConsumerMap().clear();
            this.getOtherConsumerS().stream().forEach(otherConsumer -> {
                if (this.checkMQConsumer(otherConsumer)) {
                    try {
                        DefaultLitePullConsumer defaultLitePullConsumer = this.defaultConsumer(otherConsumer);
                        String groupTopicName = otherConsumer.getGroup().concat("_").concat(otherConsumer.getTopic());
                        rocketMQTemplateS.getRocketMQTemplateMap().put(groupTopicName, this.rocketMQTemplate(rocketMQTemplateS.getDefaultMQProducer(), defaultLitePullConsumer));
                        rocketMQTemplateS.getDefaultLitePullConsumerMap().put(groupTopicName, defaultLitePullConsumer);
                    } catch (MQClientException e) {
                        log.warn("ServerName为：{}的mqEntity中[consumer]创建，请查看配置文件,异常原因为：{}", this.getServerName(), e);
                    }
                }
            });
        }
        return rocketMQTemplateS;
    }

    public RocketMQTemplate rocketMQTemplate(DefaultMQProducer defaultMQProducer, DefaultLitePullConsumer defaultLitePullConsumer) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(defaultMQProducer);
        rocketMQTemplate.setConsumer(defaultLitePullConsumer);
        RocketMQMessageConverter rocketMQMessageConverter = SpringUtil.getBean("rocketMQMessageConverter", RocketMQMessageConverter.class);
        rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        return rocketMQTemplate;
    }

    /**
     * 生成MQProducer
     *
     * @return
     */
    @Override
    public DefaultMQProducer defaultMQProducer() {
        String group = this.getProducer().getGroup();
        String ak = this.getProducer().getAccessKey();
        String sk = this.getProducer().getSecretKey();
        boolean isEnableMsgTrace = this.getProducer().isEnableMsgTrace();
        String customizedTraceTopic = this.getProducer().getCustomizedTraceTopic();
        DefaultMQProducer defaultMQProducer = RocketMQUtil.createDefaultMQProducer(group, ak, sk, isEnableMsgTrace, customizedTraceTopic);

        List<String> hostPortS = this.getHostPortS();
        defaultMQProducer.setNamesrvAddr(IPUtil.listToStr(hostPortS, ";"));

        AccessChannel accessChannel = this.getAccessChannel();
        if (ObjectUtil.ObjectUtilByLang3.isNotEmpty(accessChannel)) {
            defaultMQProducer.setAccessChannel(accessChannel);
        }
        defaultMQProducer.setInstanceName(getServerName().concat("_".concat(group)));
        defaultMQProducer.setSendMsgTimeout(this.getProducer().getSendMessageTimeout());
        defaultMQProducer.setRetryTimesWhenSendFailed(this.getProducer().getRetryTimesWhenSendFailed());
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(this.getProducer().getRetryTimesWhenSendAsyncFailed());
        defaultMQProducer.setMaxMessageSize(this.getProducer().getMaxMessageSize());
        defaultMQProducer.setCompressMsgBodyOverHowmuch(this.getProducer().getCompressMessageBodyThreshold());
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(this.getProducer().isRetryNextServer());

        if (defaultMQProducer instanceof TransactionMQProducer) {
            String threadName = this.getServerName().concat("-").concat(group);
            TransactionMQProducer transactionMQProducer = (TransactionMQProducer) defaultMQProducer;
            //设置用于事务消息的处理线程池
            ExecutorService executorService = new ThreadPoolExecutor(this.getCorePoolSize(), this.getMaximumPoolSize(),
                    this.getKeepAliveTime(), this.getUnit(), new ArrayBlockingQueue<Runnable>(this.getCapacity()), (r) -> {
                Thread thread = new Thread(r);
                thread.setName("rocketMQ-transaction-msg-check-thread-".concat(threadName));
                return thread;
            });

            transactionMQProducer.setExecutorService(executorService);
            transactionMQProducer.setTransactionListener(SpringUtil.getBean("transactionListener", TransactionListener.class));
        }

        return defaultMQProducer;
    }


    @Override
    public DefaultLitePullConsumer defaultConsumer(RocketMqConsumer mqConsumer) throws MQClientException {

        String groupName = mqConsumer.getGroup();
        String topicName = mqConsumer.getTopic();
        AccessChannel accessChannel = this.getAccessChannel();
        MessageModel messageModel = mqConsumer.getMessageModel();
        SelectorType selectorType = mqConsumer.getSelectorType();
        String selectorExpression = mqConsumer.getSelectorExpression();
        String ak = mqConsumer.getAccessKey();
        String sk = mqConsumer.getSecretKey();
        int pullBatchSize = mqConsumer.getPullBatchSize();
        List<String> hostPortS = this.getHostPortS();
        DefaultLitePullConsumer defaultLitePullConsumer
                = RocketMQUtil.createDefaultLitePullConsumer(IPUtil.listToStr(hostPortS, ";"),
                ObjectUtil.ObjectUtilByLang3.isEmpty(accessChannel) ? null : accessChannel.name(),
                groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize);
        return defaultLitePullConsumer;
    }

}
